package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo21Accumulator {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("累加器案例")

    val sc: SparkContext = new SparkContext(conf)

    //    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    //    var num = 0

    /**
     * 累加器
     * 必要有行动算子触发作业执行
     */
    //创建累加器变量
    //    val c1: LongAccumulator = sc.longAccumulator("c1")

    //    linesRDD.foreach((e: String) => {
    //      num += 1
    //      println("-----------------------")
    //      println(num)
    //    })
    //    println(s"num的值为：$num") // 0  1000

    //使用累加器
    //    val c1: LongAccumulator = sc.longAccumulator("c1")
    //    linesRDD.foreach((e:String)=>{
    //      c1.add(1)
    //    })
    //    println(s"累加之后的值为：${c1.value}")

    //使用累加器
    //    val c1: LongAccumulator = sc.longAccumulator("c1")
    //    linesRDD.map((e: String) => {
    //      c1.add(1)
    //    }).collect()
    //    println(s"累加之后的值为：${c1.value}")

    /**
     * 写spark core程序的注意事项
     * 1、RDD中无法嵌套使用RDD
     * 2、RDD中无法使用SparkContext
     */
    //    val studentLinesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //    val scoreLinesRDD: RDD[String] = sc.textFile("spark/data/score.txt")
    //
    //    val rdd1: RDD[RDD[(String, String)]] = studentLinesRDD.map((line1: String) => {
    //      scoreLinesRDD.map((line2: String) => {
    //        val s1: String = line1.split(",").mkString("|")
    //        val s2: String = line2.split(",").mkString("|")
    //        (s1, s2)
    //      })
    //    })
    //
    //    rdd1.foreach(println)

//    val studentLinesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
//    val scoreLinesRDD: RDD[String] = sc.textFile("spark/data/score.txt")
//
//    val rdd1: RDD[RDD[(String, String)]] = studentLinesRDD.map((line1: String) => {
//      sc.textFile("spark/data/score.txt").map((line2: String) => {
//        val s1: String = line1.split(",").mkString("|")
//        val s2: String = line2.split(",").mkString("|")
//        (s1, s2)
//      })
//    })


  }
}
